package com.lagou.no4



import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object No4 {
    def main(args: Array[String]): Unit = {
        //设置日志级别
        Logger.getLogger("org").setLevel(Level.WARN)
        val conf = new SparkConf()
                .setAppName(s"${this.getClass.getCanonicalName}")
                .setMaster("local[*]")
        val sc = new SparkContext(conf)
        val K = 9

        //1.转换为样例类
        val lines: RDD[LabeledPoint] = sc.textFile("data/iris.csv")
                .map{line=>
                    val fields = line.split(",").drop(1)
                    if (fields.length == 5) LabeledPoint(fields.last, fields.init.map(_.toDouble))
                    else LabeledPoint("", fields.map(_.toDouble))//没有label的将label置为空 也换为LabeledPoint样例类
                }
        //过滤掉没有类型的数据
        val sampleRDD: RDD[LabeledPoint] = lines.filter(_.label != "")
        //测试点
        val testData: Array[Array[Double]] = Array(Array(5.4,3.4,1.7,0.2),
            Array(5.1,3.7,1.5,0.4),
            Array(5.8,2.7,4.1,1.0),
            Array(5.1,2.5,3.0,1.1),
            Array(6.4,2.8,5.6,2.1),
            Array(6.7,3.1,4.7,1.5),
            Array(5.1,3.8,1.6,0.2)
        )

        //小表在外面 相当于join 小表在右边
        testData.foreach(point => {
            sampleRDD.map(labeledpoint => {
                (labeledpoint.label, getDistance(point, labeledpoint.point)) //第一个参数是已知点标签 第二个参数是已知点和未知点距离
            }).sortBy(_._2) //排序
                    .take(K) //取前9个
                    .map { case (label, _) => label } // 然后第二个参数距离就没啥用了 只取label 做WordCount
                    .groupBy(x => x)
                    .mapValues(_.length)
                    .foreach(println)
        })


        sc.stop()
    }
    import scala.math._
    def getDistance(x: Array[Double], y: Array[Double]): Double = // x y 都是有4个点的数组
        sqrt(x.zip(y).map(elem => pow(elem._1 - elem._2, 2)).sum)
    //使用zip拉链合起来 结果就像是（(x1,y1), (x2, y2),(x3, y3),(x4, y4)） 4个元组
    // 每个元组的差的平方 的 和 再sqrt开方 就是距离
}
case class LabeledPoint(label: String, point:Array[Double])